package kafka

import (
	"github.com/Shopify/sarama"
	"strings"
	"time"
)

// Config 配置
type Config struct {
	TopicA     string `xml:"topica"`
	TopicB     string `xml:"topicb"`
	Broker     string `xml:"broker"`
	Frequency  int    `xml:"frequency"`
	MaxMessage int    `xml:"max_message"`
}

// NewProducer 构造KafkaProducer
func NewProducer(cfg *Config) (*Producer, error) {

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.NoResponse                                  // Only wait for the leader to ack
	config.Producer.Compression = sarama.CompressionSnappy                            // Compress messages
	config.Producer.Flush.Frequency = time.Duration(cfg.Frequency) * time.Millisecond // Flush batches every 500ms
	config.Producer.Partitioner = sarama.NewRandomPartitioner

	p, err := sarama.NewAsyncProducer(strings.Split(cfg.Broker, ","), config)
	if err != nil {
		return nil, err
	}
	ret := &Producer{
		producer:  p,
		topica:  cfg.TopicA,
		topicb:  cfg.TopicB,
		closeChan: make(chan struct{}),
	}
	return ret, nil
}
